package com.qxp.simplerpc;


import com.qxp.simplerpc.config.RedisAutoConfiguration;
import com.qxp.simplerpc.entity.CommonProperties;
import com.qxp.simplerpc.parse.URLParse;
import com.qxp.simplerpc.utils.RedisUtils;
import com.qxp.simplerpc.utils.SimpleRpcProperties;
import com.qxp.simplerpc.connect.RpcClientHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Import;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Component
@Import({RedisAutoConfiguration.class,RedisUtils.class})
public class RedisRegistryServiceImpl extends AbstractRegistryService {

    @Autowired
    private RedisUtils redisUtils;

    private static final long timeout = 10L;
    @Autowired
    public void setSimpleRpcProperties(SimpleRpcProperties simpleRpcProperties) {
        this.simpleRpcProperties = simpleRpcProperties;
    }
    @PostConstruct
    public void initRegister() {
        initRegisterProperties();
    }

    @Override
    public void register(String url) {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleWithFixedDelay(() -> doRegister(url), timeout, timeout, TimeUnit.SECONDS);
    }

    protected void doRegister(String url) {
        String key = stringFormatWithSplit(clustername, nodeName);
        clearUnusedRegisters(clustername);
        redisUtils.addMap2Value(clustername, key, url, timeout, TimeUnit.SECONDS);
        redisUtils.set(key, url, timeout, TimeUnit.SECONDS);
    }

    @Override
    public void unregister(String url) {
        redisUtils.removeMap4Value(clustername, nodeName);
    }

    @Override
    public void subscribe(RpcClientHandler rpcClientHandler) {
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(() -> doSubscribe(rpcClientHandler), timeout, timeout, TimeUnit.SECONDS);
    }

    @Override
    public String unsubcribe(RpcClientHandler rpcClientHandler) {
        return null;
    }

    @Override
    public List<String> listUrl() {
        List<String> list = new ArrayList<>();
        Map<String, Object> result = redisUtils.get(clustername);
        result.values().forEach(res -> {
            if (!Objects.isNull(res)) {
                list.add(res.toString());
            }
        });
        return list;
    }

    private void clearUnusedRegisters(String key) {
        Map<String, Object> registers = redisUtils.get(key);
        if (CollectionUtils.isEmpty(registers)) {
            return;
        }
        Map<String, Object> res = new HashMap<>();
        registers.forEach((k, v) -> {
            if (!v.equals(redisUtils.get(k))) {
                res.put(k, v);
            }
        });
        redisUtils.removeAllMap2Value(key, res, timeout, TimeUnit.SECONDS);
    }

}
